package com.atguigu.flink.chapter11;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

public class Flink13_Window_Grouped_SQL {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

               //因为加了window 后变成普通的流的，不是source？？？！！！ 不是 DataStreamSource 了
           DataStream<WaterSensor> waterSensorStream =
                   env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                           new WaterSensor("sensor_1", 2000L, 20),
                           new WaterSensor("sensor_2", 3000L, 30),
                           new WaterSensor("sensor_1", 4000L, 40),
                           new WaterSensor("sensor_1", 3000L, 50),
                           new WaterSensor("sensor_2", 6000L, 60))
                   .assignTimestampsAndWatermarks(
                           WatermarkStrategy
                                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                   //事件事件来自于时间戳
                                   .withTimestampAssigner((event, timestamp) -> event.getTs())
                   );


           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //  每隔5s 统计最近5s 内 每个传感器的水位和
           Table table = tableEnv.fromDataStream(waterSensorStream, $("id"), $("ts").rowtime(), $("vc"));

           //注册表
           tableEnv.createTemporaryView("sensor",table);

           // 滚动窗口
          /* tableEnv.sqlQuery(" select " +
                   " id ," +
                   " Tumble_start(ts,interval '5' second) stt," +  // 滚动的开始时间
                   " Tumble_end(ts,interval '5' second) edt," +  // 滚动的结束时间
                   "  sum(vc) sum_vc" +
                   "  from sensor " +
                   " group by id ,TUMBLE(ts,interval '5' second)")
                   .execute()
                   .print();*/

           // 滑动窗口

          /* tableEnv.sqlQuery(" select " +
                   " id ," +
                   " Hop_start(ts,interval '2' second,interval '5' second) stt," +  // 滑动的开始时间
                   " Hop_end(ts,interval '2' second,interval '5' second) edt," +  // 滑动的开始时间
                   "  sum(vc) sum_vc" +
                   "  from sensor " +
                   " group by id ,hop(ts,interval '2' second,interval '5' second)")
                   .execute()
                   .print();*/

           // 会话窗口
           tableEnv.sqlQuery(" select " +
                   " id ," +
                   " SESSION_start(ts,interval '2' second) stt," +  // 滑动的开始时间
                   " SESSION_end(ts,interval '2' second) edt," +  // 滑动的开始时间
                   "  sum(vc) sum_vc" +
                   "  from sensor " +
                   " group by id ,SESSION(ts,interval '2' second)")
                   .execute()
                   .print();



       }
}
